package com.dianmi.seckill.service.impl;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.dianmi.seckill.common.advice.ServiceException;
import com.dianmi.seckill.common.utils.FastUtils;
import com.dianmi.seckill.common.utils.JsonUtil;
import com.dianmi.seckill.dao.ProductDao;
import com.dianmi.seckill.entity.ProductStockFlag;
import com.dianmi.seckill.entity.SeckillProduct;
import com.dianmi.seckill.service.ProductService;
import com.dianmi.seckill.service.StockCoordinate;
import com.dianmi.seckill.vo.SeckillProductInfo;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author xieqiaofu
 * @date 2021/4/2
 */
@Slf4j
@Service
public class ProductServiceImpl implements ProductService {

    private static final String TRANS_SUBSTRACT_FAILED_MSG = "transcation substract failed, productId: %s, userId: %s";

    private static final String PRODUCT_LIST_KEY = "seckill_product_list";

    private static final String PRODUCT_STOCK_PREFIX = "product_stock_prefix:";

    private static final String REDUCE =
            "local stock = redis.call(\"DECR\", KEYS[1]) " +
            "if tonumber(stock) < tonumber(ARGV[1]) " +
            "then " +
            "   redis.call(\"INCR\", KEYS[1]) " +
            "   return 0 " +
            "else " +
            "   return 1 " +
            "end ";

    //各个集群节点会通过监听zk达到一致
    private static ConcurrentHashMap<Long, Boolean> stockFlags = new ConcurrentHashMap();

    @Resource
    private ProductDao productDao;

    @Autowired
    private StockCoordinate stockCoordinate;

    @Autowired
    private RedisTemplate redisTemplate;

    private StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

    @PostConstruct
    public void init() {
        stockCoordinate.createBaseNode();

        /*初始化zk商品信息到内存*/
        initProductStock();


        /*监听节点变化*/
        stockCoordinate.listen(event -> {
            if (PathChildrenCacheEvent.Type.CHILD_UPDATED == event.getType()) {
                String path = event.getData().getPath();
                Long productId = Long.valueOf(path.substring(path.lastIndexOf("/") + 1));
                Integer data = Integer.valueOf(new String(event.getData().getData(), StandardCharsets.UTF_8));
                stockFlags.put(productId, data == 1);
            }
        });
    }

    private void initProductStock() {
        List<ProductStockFlag> flags = stockCoordinate.getProductStockFlag();
        if (CollectionUtils.isEmpty(flags)) {
            return;
        }
        for (ProductStockFlag flag : flags) {
            stockFlags.put(flag.getProductId(), String.valueOf(1).equals(flag.getData()));
        }
    }

    @Override
    public Boolean upToShelves(SeckillProductInfo productInfo) {
        /*1、保存秒杀商品信息*/
        productDao.insert(FastUtils.convertCopy(SeckillProduct.class, productInfo));
        /*2、注册商品节点到zk*/
        stockCoordinate.register(productInfo.getId(), toMilli(productInfo.getEndTime()) - toMilli(productInfo.getStartTime()));
        /*3、做redis缓存*/
        /*3-1、缓存商品基本信息，查询商品列表用*/
        cacheProductBaseInfo(productInfo);
        /*3-2、缓存库存信息*/
        cacheProductStockInfo(productInfo.getId(), productInfo.getStock());
        /*4、生成库存剩余标志*/
        generateStockFlag(productInfo.getId());
        return true;
    }

    /**
     * 查询商品库存标志
     * @param productId
     * @return
     */
    public Boolean stockFlag(Long productId) {
        Boolean flag = stockFlags.get(productId);
        if (null == flag) {
            throw new ServiceException(500, "商品异常");
        }
        return flag;
    }

    /**
     * 修改商品库存标志
     * @param productId
     * @param flag
     */
    public void updateStockFlag(Long productId, Boolean flag) {
        stockFlags.put(productId, flag);
    }

    /**
     * redis扣减库存
     * @param productId
     * @return
     */
    public Boolean reduce(Long productId) {
        RedisScript<Long> script = new DefaultRedisScript<>(REDUCE, Long.class);
        Long ret = (Long)redisTemplate.execute(script, stringRedisSerializer, stringRedisSerializer, Lists.newArrayList(PRODUCT_STOCK_PREFIX + productId), "0");
        if (null != ret && Long.valueOf(1L).equals(ret)) {
            return true;
        }
        return false;
    }

    /**
     * 还原redis库存
     * @param productId
     */
    public void restore(Long productId) {
        redisTemplate.opsForValue().increment(PRODUCT_STOCK_PREFIX + productId);
        updateStockFlag(productId, Boolean.FALSE);
        stockCoordinate.update(productId, true);
    }

    public SeckillProduct getProductInfo(Long productId) {
        List<String> range = (List<String>)redisTemplate.opsForList().range(PRODUCT_LIST_KEY, 0, -1);
        if (CollectionUtils.isEmpty(range)) {
            List<SeckillProduct> list = productDao.selectList(Wrappers.<SeckillProduct>lambdaQuery().orderByDesc(SeckillProduct::getCreateTime));
            if (CollectionUtils.isEmpty(range)) {
                return null;
            }
            List<String> strings = new ArrayList<>();
            for (SeckillProduct pro : list) {
                strings.add(JsonUtil.toJson(pro));
            }
            range = strings;
            redisTemplate.opsForList().leftPushAll(PRODUCT_LIST_KEY, strings);
        }
        for (String str : range) {
            SeckillProduct product = JsonUtil.parseJson(str, SeckillProduct.class);
            if (product.getId().equals(productId)) {
                return product;
            }
        }
        return null;
    }

    private void generateStockFlag(Long productId) {
        stockFlags.putIfAbsent(productId, Boolean.TRUE);
    }

    private void cacheProductStockInfo(Long productId, Integer stock) {
        redisTemplate.opsForValue().set(PRODUCT_STOCK_PREFIX + productId, String.valueOf(stock));
    }

    private void cacheProductBaseInfo(SeckillProductInfo productInfo) {
        SeckillProduct product = FastUtils.convertCopy(SeckillProduct.class, productInfo);
        redisTemplate.opsForList().leftPush(PRODUCT_LIST_KEY, JsonUtil.toJson(product));
    }

    private Long toMilli(LocalDateTime time) {
        return time.toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
    }
}
